package org.qiyu.live.user.provider.service.impl;


import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.qiyu.live.common.interfaces.utils.ConvertBeanUtils;
import org.qiyu.live.user.interfaces.dto.UserDTO;
import org.qiyu.live.user.provider.dao.mapper.IUserMapper;
import org.qiyu.live.user.provider.dao.po.UserPO;
import org.qiyu.live.user.provider.service.IUserService;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class UserServiceImpl implements IUserService {

    @Resource
    private IUserMapper userMapper;

    @Resource
    private RedisTemplate<String,UserDTO> softwareRedisTemplate;

    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;

    @Resource
    private MQProducer mqProducer;

    @Override
    public UserDTO getByUserId(Long userId) {
        if(userId == null){
            return null;
        }

        //qiyu-live-user-application:userInfo:10001
        String redisKey = cacheKeyBuilder.buildUserInfoKey(userId);
        UserDTO userDTO = softwareRedisTemplate.opsForValue().get(redisKey);

        if(userDTO!=null){
            return userDTO;
        }

        UserPO userPO = userMapper.selectById(userId);

        userDTO = ConvertBeanUtils.convert(userPO, UserDTO.class);

        if(userDTO != null){
            softwareRedisTemplate.opsForValue().set(redisKey,userDTO,30, TimeUnit.MINUTES);
        }
        return userDTO;
    }


    @Override
    public boolean insertOne(UserDTO userDTO) {
        //1、校验参数
        if(userDTO== null || userDTO.getUserId()==null){
            return false;
        }
        //2、userDTO 转userPO
        UserPO userPO = ConvertBeanUtils.convert(userDTO, UserPO.class);
        userMapper.insert(userPO);
        return true;
    }

    @Override
    public boolean updateUserInfo(UserDTO userDTO) {
        //1、校验参数
        if(userDTO ==null || userDTO.getUserId()==null){
            return false;
        }
        UserPO userPO = ConvertBeanUtils.convert(userDTO, UserPO.class);
        userMapper.updateById(userPO);

        String key = cacheKeyBuilder.buildUserInfoKey(userDTO.getUserId());
        //立即删除缓存
        softwareRedisTemplate.delete(key);
        try {
            //发送延迟消息，触发二次删除缓存操作
            Message message = new Message();
            message.setBody(JSON.toJSONString(userDTO).getBytes());
            message.setTopic("user-update-cache");
            //延迟级别，1代表延迟一秒发送
            message.setDelayTimeLevel(1);
            mqProducer.send(message);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return true;
    }






    @Override
    public Map<Long, UserDTO> batchQueryUserInfo(List<Long> userIdList) {

        //1、校验参数
        if(CollectionUtils.isEmpty(userIdList)){
           return new HashMap<>();
        }
        //2、利用stream流过滤出用户id大于10000的数据
        userIdList = userIdList.stream().filter(x -> x > 10000).collect(Collectors.toList());

        //3、key的集合
        List<String> keyList = new ArrayList<>();
        for (Long userId : userIdList) {
            keyList.add(cacheKeyBuilder.buildUserInfoKey(userId));
        }
        //4、从redis中获取缓存的用户数据
        List<UserDTO> userDTOList = softwareRedisTemplate.opsForValue().multiGet(keyList).stream().filter(x->x!=null).collect(Collectors.toList());

        if(!CollectionUtils.isEmpty(userDTOList)&&userDTOList.size()==userIdList.size()){
            //所有的缓存都命中了，直接将数据返回给前端
            return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId,x->x));
        }
        //得到redis缓存中用户的id的集合
        List<Long> userIdInCacheList = userDTOList.stream().map(UserDTO::getUserId).collect(Collectors.toList());
        //得到没有命中缓存的用户的id集合
        List<Long> userIdNotInCacheList = userIdList.stream().filter(x -> !userIdInCacheList.contains(x)).collect(Collectors.toList());
        //{100001,100002,100003,100004,100101，100102,110102}
        //     01     02     03     04     01      02    02
        //{100001,100101}   {100002,100102,110102}, {100003}, {100004}
        Map<Long, List<Long>> userIdMap = userIdNotInCacheList.stream().collect(Collectors.groupingBy(x -> x % 100));

        List<UserDTO> dbQueryResult = new CopyOnWriteArrayList<>();
        // 采用多线程+本地内存的方法实现批量用户查询
        userIdMap.values().parallelStream().forEach(queryUserIdList->{
            dbQueryResult.addAll(ConvertBeanUtils.convertList(userMapper.selectBatchIds(queryUserIdList),UserDTO.class));
        });

        if(!CollectionUtils.isEmpty(dbQueryResult)){
            Map<String, UserDTO> saveCacheMap = dbQueryResult.stream().collect(Collectors.toMap(userDTO -> cacheKeyBuilder.buildUserInfoKey(userDTO.getUserId()), x -> x));
            softwareRedisTemplate.opsForValue().multiSet(saveCacheMap);
            softwareRedisTemplate.executePipelined(new SessionCallback<Object>() {
                @Override
                public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                    for (String key : saveCacheMap.keySet()) {
                        operations.expire((K) key,createRandomTime(),TimeUnit.SECONDS);
                    }
                    return null;
                }
            });

            userDTOList.addAll(dbQueryResult);
        }

        return userDTOList.stream().collect(Collectors.toMap(UserDTO::getUserId,x->x));
    }

    //创建一个随机的数字
    private int createRandomTime(){
        return ThreadLocalRandom.current().nextInt(100)+30*60;
    }


}
